package org.example;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class PaimonDDL {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setString(RestOptions.BIND_PORT,"8081");
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment(conf);
        env.setParallelism(1);
        TableEnvironment tableEnv = StreamTableEnvironment.create(env);

        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);

        tableEnv.executeSql("CREATE CATALOG paimon WITH (\n" +
                "    'type' = 'paimon',\n" +
                "    'warehouse' = 'file:///tmp/paimon'\n" +
                ");");

        tableEnv.executeSql("ALTER TABLE `paimon`.`default`.`test` RESET ('write-buffer-size');");

        tableEnv.executeSql("show create table `paimon`.`default`.`test` ").print();

//        tableEnv.executeSql("DROP TABLE IF EXISTS `paimon`.`default`.`test`;");

//        tableEnv.executeSql("CREATE TABLE IF NOT EXISTS `paimon`.`default`.`test` (\n" +
//                "                user_id BIGINT,\n" +
//                "                item_id BIGINT,\n" +
//                "                behavior STRING,\n" +
//                "                dt STRING,\n" +
//                "                hh STRING,\n" +
//                "                PRIMARY KEY (dt, hh, user_id) NOT ENFORCED\n" +
//                "                );");
//
//        tableEnv.executeSql("CREATE TEMPORARY TABLE temp_table (\n" +
//                "    user_id INT,\n" +
//                "    behavior STRING\n" +
//                ") WITH (\n" +
//                "    'connector' = 'filesystem',\n" +
//                "    'path' = '/tmp/test/temp_table.csv',\n" +
//                "    'format' = 'csv'\n" +
//                ");");

//        tableEnv.executeSql("INSERT INTO `paimon`.`default`.`test`\n" +
//                "SELECT\n" +
//                "    t1.user_id\n" +
//                "    , 1 as item_id\n" +
//                "    , t1.behavior as behavior\n" +
//                "    , '2025-01-07' as dt\n" +
//                "    , '11' as hh\n" +
//                "FROM temp_table t1\n" +
//                ";");

        // 读取Paimon表数据
//        tableEnv.executeSql("SELECT t1.*\n" +
//                "FROM `paimon`.`default`.`test` t1\n" +
//                "JOIN temp_table ON t1.user_id = temp_table.user_id;\n").print();
    }
}
